package com.alipay.sofa.demo.db.provider.producer;

import com.alibaba.fastjson.JSONObject;
import com.alipay.sofa.demo.db.common.entity.StudentInfo;
import com.alipay.sofa.demo.db.common.entity.StudentInfoDTO;
import com.alipay.sofa.demo.db.common.tool.DataMockUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 定时发送插入学生数据消息类
 * （定时轮询插入状态，如果连续maxDelay分钟，没有插入成功，进行重试）
 * @Author Zhang Da
 * @Date 2020/2/15 19:42
 * @Version 1.0
 */

@Component
public class StudentInfoProducer {
    @Resource
    private KafkaTemplate kafkaTemplate;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private RedisClient redisClient;

    /**
     * 消息消费情况超期时间（单位：ms）
     * （如果超过超期时间3分钟，将会发起重试）
     */
    private static final int EXPIRE_TIE = 300000;
    /**
     * 核心线程数
     */
    private static final int THREAD_COUNT = 32;
    /**
     * 是否停止
     */
    private static boolean stop = false;
    /**
     * 定时发送任务执行次数
     */
    private static AtomicInteger scheduledTaskCount  = new AtomicInteger();
    /**
     * 发送请求总数量
     */
    private static AtomicInteger totalCount  = new AtomicInteger();
    /**
     * 发送请求计数：成功数
     */
    private static AtomicInteger sendSuccessTotal = new AtomicInteger();
    /**
     * 发送请求计数：失败数
     */
    private static AtomicInteger sendFailTotal = new AtomicInteger();

    /**
     * 消费计数：成功数
     */
    private static AtomicInteger consumeSuccessTotal = new AtomicInteger();
    /**
     * 消费计数：失败数
     */
    private static AtomicInteger consumeFailTotal = new AtomicInteger();
    /**
     * 格式化日志格式
     */
    private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private static final Logger logger = LoggerFactory.getLogger(StudentInfoProducer.class);

    private static long startTime;

    static {
        startTime = System.currentTimeMillis();
    }

    /**
     * 定义线程池，用于并发发送数据
     */
    private ExecutorService pool = new ThreadPoolExecutor(THREAD_COUNT, THREAD_COUNT*20, 300, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(THREAD_COUNT), Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());

    @Scheduled(fixedDelay = 6000000)
    public void executeSendMsg(){
        pool.execute(()-> {
            while (!stop) {
                scheduledTaskCount.addAndGet(1);
                sendMsg(generateStudentDTO());
            }
        });
        pool.shutdown();
//        scheduledTaskCount.addAndGet(1);
//        sendMsg(generateStudentDTO());

    }

    /**
     * 定期校验插入操作是否成功
     */
//    @Scheduled(fixedDelay = 10000)
//    public void executeVerify() {
//        Set<String> keys =  stringRedisTemplate.keys("*");
//        Iterator<String> it1 = keys.iterator();
//        //遍历所有以prefix_message开头的key
//        while (it1.hasNext()) {
//            String key = it1.next();
//            if(key != null && !"".equals(key.trim())){
//                StudentInfoDTO studentInfoDTO = redisClient.getByProtostuff(key, StudentInfoDTO.class);
//                if(studentInfoDTO != null && studentInfoDTO.isConsumeSuccess()){
//                    //如果已经插入成功，则删除该条记录
//                    consumeSuccessTotal.getAndAdd(1);
//                    redisClient.delete(key);
////                    logger.info("消息《消费》成功，messageId：" + key);
//                }else{
//                    //如果尚未消费成功，拿到当时发送的消息value
//                    StudentInfoDTO sid = redisClient.getByProtostuff(key, StudentInfoDTO.class);
//                    //拿到消息发送时间
//                    long sendTime = sid.getSendTime();
//                    //将消息发送时间跟超时时间进行比对
//                    if(System.currentTimeMillis() - sendTime > EXPIRE_TIE){
////                        logger.error(String.format("消息《消费》失败，messageId：%s，现发起重试", key));
//                        consumeFailTotal.getAndAdd(1);
//                        sid.setSendTime(System.currentTimeMillis());
//                        //重试
//                        sendMsg(sid);
//                        //删除redis中的信息，重试后会重新写入
//                        redisClient.delete(key);
//                    }
//
//                }
//            }
//        }
//    }

    /**
     * 定期输出任务执行的统计信息
     */
    @Scheduled(fixedDelay = 5000)
    public void executePrintStatistics() {
        long period = (System.currentTimeMillis() - startTime)/1000;
        logger.info("--->任务统计信息输出开始，当前时间：" + simpleDateFormat.format(new Date()));
        logger.info("--->消息《发送》总次数：" + totalCount.get());
        logger.info("--->消息《发送》成功数：" + sendSuccessTotal.get());
        logger.info("--->消息《发送》失败数：" + sendFailTotal.get());
        logger.info("--->发送消息qps为：" + sendSuccessTotal.get()/period);
        logger.info("--->消息《消费》成功数：" + consumeSuccessTotal.get());
        logger.info("--->消息《消费》失败数：" + consumeFailTotal.get());

    }

    /**
     * 发送一条消息，消息内容为StudentInfoDTO
     */
    public void sendMsg(StudentInfoDTO studentInfoDTO){
        //将数据存储在redis中，定时进行校验是否消费成功，如果消费成功，则会删除数据，否则，超时会进行重发
        redisClient.setByProtostuff(studentInfoDTO.getMessageId(), studentInfoDTO, StudentInfoDTO.class);

        ListenableFuture<SendResult<Integer, String>> listenableFuture =
                kafkaTemplate.send("insertStu",  studentInfoDTO);
        totalCount.addAndGet(1);
        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            @Override
            public void onSuccess(SendResult<Integer, String> result) {
                sendSuccessTotal.addAndGet(1);
//                logger.info("消息《发送》成功，messageId：" + studentInfoDTO.getMessageId());
            }

            @Override
            public void onFailure(Throwable ex) {
                sendFailTotal.addAndGet(1);
                logger.error("消息《发送》失败，messageId：" + studentInfoDTO.getMessageId());
            }
        });
    }

    /**
     * 随机生成包含一或多条学生信息的StudentInfoDTO
     * @return
     */
    public StudentInfoDTO generateStudentDTO(){
        List<StudentInfo> list = new ArrayList<StudentInfo>();
        int randomStudentNumber = DataMockUtil.getDistanceRandomNum(1, 10);
        for(int i=0; i<randomStudentNumber; i++){
            list.add(DataMockUtil.mockOneStudent());
        }
        StudentInfoDTO studentInfoDTO = new StudentInfoDTO();
        studentInfoDTO.setMessageId(UUID.randomUUID().toString());
        studentInfoDTO.setSendTime(System.currentTimeMillis());
        studentInfoDTO.setStudentInfoList(list);
        studentInfoDTO.setConsumeSuccess(false);
        return studentInfoDTO;
    }

    public int getTotalCount() {
        return totalCount.get();
    }

}
